import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * kafka 读取数据到 mysql中
 */
public class kafkaTableTest {
    public static void main(String[] args) throws Exception {
        System.out.printf("hello word");

        // Kafka {"msg": "welcome flink users..."}
        String sourceDDL = "CREATE TABLE kafka_source (\n" +
                " msg STRING\n" +
                ") WITH (\n" +
                " 'connector' = 'kafka-0.11',\n" +
                " 'topic' = 'cdn_log',\n" +
                " 'properties.bootstrap.servers' = '192.168.235.128:9092',\n" +
                " 'format' = 'json',\n" +
                " 'scan.startup.mode' = 'latest-offset'\n" +
                ")";

        // Mysql
        String sinkDDL = "CREATE TABLE mysql_sink (\n" +
                " msg STRING \n" +
                ") WITH (\n" +
                "  'connector' = 'jdbc',\n" +
                "   'url' = 'jdbc:mysql://192.168.235.128:3306/flinkdb?characterEncoding=utf-8&useSSL=false',\n" +
                "   'table-name' = 'cdn_log',\n" +
                "   'username' = 'root',\n" +
                "   'password' = '123456',\n" +
                "   'sink.buffer-flush.max-rows' = '1'\n" +
                ")";
        //print
        String printDDL = "CREATE TABLE print_sink (\n" +
                " msg STRING \n" +
                ") WITH (\n" +
                " 'connector' = 'print'\n" +
                ")";
//        EnvironmentSettings environmentSettings= EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
//        TableEnvironment tableEnvironment=TableEnvironment.create(environmentSettings);
        StreamExecutionEnvironment bsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings=EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(bsEnv, settings);
        bsTableEnv.executeSql(sourceDDL);
        bsTableEnv.executeSql(sinkDDL);
//        bsTableEnv.executeSql(printDDL);
      Table table= bsTableEnv.from("kafka_source");
        table.insertInto("mysql_sink");
//          table.insertInto("print_sink");

        bsTableEnv.execute("hello flink data");
    }
}
